package kafka

import (
	"context"
	"github.com/Sir-yuChen/funAnalyse/global"
	"github.com/Sir-yuChen/funAnalyse/utils/kafka/spec"
	"go.uber.org/zap"
	"strconv"
)

func GetKafkaProducer(link string) (*Producer, error) {
	producer, err := NewProducer(
		link,
		WithProducerErrorHandler(func(err *AsyncProduceError) {
			//todo 异步消息发送异常，处理
			global.GVA_LOG.Error("kafka-->异步生产消息异常", zap.Error(err.Err), zap.Any("msg", err.Msg))
		}), WithProducerAckMode(spec.WaitLeader))
	if err != nil {
		global.GVA_LOG.Error("kafka-->异步生产消息异常", zap.Error(err))
		return producer, err
	}
	return producer, err
}

func SendMsgToKafka(producer *Producer, topic, key string, sendContent string, seq int) (err error) {
	size, err := producer.SyncProduce(topic, key, MsgDataRequest{
		Content: sendContent,
		Seq:     strconv.Itoa(seq),
	})
	if err != nil {
		global.GVA_LOG.Error("kafka推送数据异常",
			zap.Any("Msgkey", key), zap.Any("Msg", sendContent), zap.Any("Seq", seq),
			zap.Any("数据大小", size), zap.Error(err))
		return
	} else {
		global.GVA_LOG.Info("消息发送成功!", zap.Any("发送条数", seq), zap.Any("key", key), zap.Any("MsgContent", sendContent))
		seq++
	}
	return
}

/*
	queueExchange 交换机
	receiver   执行类
	runNums  开启并发执行任务数量
	消费组：
		consumer group下可以有一个或多个consumer instance，consumer instance可以是一个进程，也可以是一个线程
		group.id是一个字符串，唯一标识一个consumer group
		consumer group下订阅的topic下的每个分区只能分配给某个group下的一个consumer(当然该分区还可以被分配给其他group)
*/

func KafkaRecv(brokers string, topics []string, groupName string, kafkaConsumer KafkaConsumer, runNums int) {
	consumersChannl := make(chan Consumer)
	var consumers []spec.Consumer
	for i := 1; i <= runNums; i++ {
		go func() {
			var consumer spec.Consumer
			var err error
			consumer, err = NewConsumer(
				brokers,
				topics,
				groupName,
				kafkaConsumer.Consumer,
				WithConsumerErrorHandler(kafkaConsumer.FailAction),
				WithConsumerAOR(spec.AORMode(global.GVA_CONFIG.KafkaConfig.AORMode)), //Newest  Earliest
				WithConsumerManualCommit(true))
			if err != nil {
				global.GVA_LOG.Error("Kafka 异常", zap.Any("err", err))
			}
			consumer.StartBlock(context.Background())
			consumers = append(consumers, consumer)
		}()
	}
	defer func() {
		for _, consumer := range consumers {
			if err := consumer.Close(); err != nil {
				global.GVA_LOG.Error("Kafka  关闭消费者异常", zap.Any("err", err))
			}
		}
	}()
	<-consumersChannl
	global.GVA_LOG.Error("Kafka  阻塞失败")
}
